package net.qiqbframework.commandhandling;

import lombok.extern.slf4j.Slf4j;
import net.qiqbframework.common.Registration;
import net.qiqbframework.common.transaction.TransactionManager;
import net.qiqbframework.config.Configuration;
import net.qiqbframework.messaging.*;
import net.qiqbframework.messaging.intercepting.MessageDispatchInterceptor;
import net.qiqbframework.messaging.monitoring.MessageMonitor;
import net.qiqbframework.messaging.monitoring.MonitorCallback;
import net.qiqbframework.messaging.task.BatchingMessageTask;
import net.qiqbframework.messaging.task.DefaultMessageTask;
import net.qiqbframework.messaging.task.MessageTask;
import net.qiqbframework.messaging.task.RollbackConfiguration;
import net.qiqbframework.tracing.Span;

import javax.annotation.Nonnull;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;

import static net.qiqbframework.commandhandling.GenericCommandResultMessage.asCommandResultMessage;
import static net.qiqbframework.common.BuilderUtils.assertNonNull;


@Slf4j
public class DefaultCommandBus implements CommandBus {


    private final TransactionManager transactionManager;
    private final MessageMonitor<? super CommandMessage<?>> messageMonitor;
    private final DuplicateCommandHandlerResolver duplicateCommandHandlerResolver;
    private final ConcurrentMap<String, CommandMessageHandler> subscriptions = new ConcurrentHashMap<>();
    private final List<MessageHandlerInterceptor<? super CommandMessage<?>>> handlerInterceptors = new CopyOnWriteArrayList<>();
    private final List<MessageDispatchInterceptor<? super CommandMessage<?>>> dispatchInterceptors = new CopyOnWriteArrayList<>();
    private final CommandCallback<Object, Object> defaultCommandCallback;
    private RollbackConfiguration rollbackConfiguration;
    private final CommandBusSpanFactory spanFactory;
    private final Configuration configuration;

    protected DefaultCommandBus(DefaultCommandBusBuilder builder) {
        builder.validate();
        this.configuration = builder.configuration;
        this.transactionManager = builder.transactionManager;
        this.messageMonitor = builder.messageMonitor;
        this.rollbackConfiguration = builder.rollbackConfiguration;
        this.duplicateCommandHandlerResolver = builder.duplicateCommandHandlerResolver;
        this.defaultCommandCallback = builder.defaultCommandCallback;
        this.spanFactory = builder.builderSpanFactory;
    }


    /**
     * 调用所有拦截器
     *
     * @param command The original command being dispatched
     * @param <C>     The type of payload contained in the CommandMessage
     * @return The command to actually dispatch
     */
    @SuppressWarnings("unchecked")
    protected <C> CommandMessage<C> intercept(CommandMessage<C> command) {
        CommandMessage<C> commandToDispatch = command;
        for (MessageDispatchInterceptor<? super CommandMessage<?>> interceptor : dispatchInterceptors) {
            commandToDispatch = (CommandMessage<C>) interceptor.handle(commandToDispatch);
        }
        return commandToDispatch;
    }


    protected <C, R> void doDispatch(MessageTask<CommandMessage<?>> messageMessageTask, CommandMessage<C> command, CommandCallback<? super C, ? super R> callback) {
        MonitorCallback monitorCallback = messageMonitor.onMessageIngested(command);
        Optional<MessageHandler<? super CommandMessage<?>>> optionalHandler = findCommandHandlerFor(command);
        if (optionalHandler.isPresent()) {
            CommandMessage<C> commandWithContext = spanFactory.propagateContext(command);
            handle(messageMessageTask, commandWithContext, optionalHandler.get(), new MonitorAwareCallback<>(callback, monitorCallback));
        } else {
            CommandHandlerNotFoundException exception = new CommandHandlerNotFoundException("No handler was subscribed for command [%s].".formatted(command.getCommandName()));
            monitorCallback.reportFailure(exception);
            callback.onResult(command, asCommandResultMessage(exception));
        }

    }

    private Optional<MessageHandler<? super CommandMessage<?>>> findCommandHandlerFor(CommandMessage<?> command) {
        final String commandName = command.getCommandName();
        return Optional.ofNullable(subscriptions.get(commandName));
    }


    protected <C, R> void handle(MessageTask<CommandMessage<?>> messageMessageTask, CommandMessage<C> command,
                                 MessageHandler<? super CommandMessage<?>> handler,
                                 CommandCallback<? super C, ? super R> callback) {
        CommandResultMessage<R> resultMessage = spanFactory.createHandleCommandSpan(command).runSupplier(() -> {
            log.info("Handling command [{}]", command.getCommandName());
            InterceptorChain chain = new DefaultInterceptorChain<>(messageMessageTask, handlerInterceptors, handler);
            try {
                return asCommandResultMessage(chain.proceed());
            } catch (Exception e) {
                if (e instanceof RuntimeException runtimeException){
                    throw runtimeException;
                }
                throw new RuntimeException(e);
            }
        });
        callback.onResult(command, resultMessage);
    }

    @Override
    public <C> void dispatch(@Nonnull CommandMessage<C> command) {
        dispatch(command, defaultCommandCallback);
    }

    @Override
    public <C, R> void dispatch(@Nonnull CommandMessage<C> command, @Nonnull CommandCallback<? super C, ? super R> callback) {

        Span span = spanFactory.createDispatchCommandSpan(command);
        span.run(() -> {
            CommandCallback<? super C, ? super R> spanAwareCallback = callback.wrap((commandMessage, commandResultMessage) -> {
                if (commandResultMessage.isExceptional()) {
                    span.recordException(commandResultMessage.exceptionResult());
                }
            });
            MonitorCallback monitorCallback = messageMonitor.onMessageIngested(command);
            Optional<MessageHandler<? super CommandMessage<?>>> optionalHandler = findCommandHandlerFor(command);
            if (optionalHandler.isPresent()) {
                CommandMessage<C> commandWithContext = spanFactory.propagateContext(command);
                CommandResultMessage<R> resultMessage = spanFactory.createHandleCommandSpan(commandWithContext).runSupplier(() -> {
                    log.debug("Handling command [{}]", commandWithContext.getCommandName());
                    MessageTask<CommandMessage<?>> messageMessageTask = DefaultMessageTask.startAndGet(commandWithContext);
                    messageMessageTask.attachTransaction(transactionManager);
                    InterceptorChain chain = new DefaultInterceptorChain<>(messageMessageTask, handlerInterceptors, optionalHandler.get());
                    return asCommandResultMessage(messageMessageTask.executeWithResult(chain::proceed,
                            rollbackConfiguration));
                });
                spanAwareCallback.onResult(command, resultMessage);
            } else {
                CommandHandlerNotFoundException exception = new CommandHandlerNotFoundException("No handler was subscribed for command [%s].".formatted(command.getCommandName()));
                monitorCallback.reportFailure(exception);
                spanAwareCallback.onResult(command, asCommandResultMessage(exception));
            }
        });

    }


    public <C, R> void dispatch(MessageTask<CommandMessage<?>> messageMessageTask, @Nonnull CommandMessage<C> command, @Nonnull CommandCallback<? super C, ? super R> callback) {

        Span span = spanFactory.createDispatchCommandSpan(command);
        span.run(() -> {
            CommandCallback<? super C, ? super R> spanAwareCallback = callback.wrap((commandMessage, commandResultMessage) -> {
                if (commandResultMessage.isExceptional()) {
                    span.recordException(commandResultMessage.exceptionResult());
                }
            });
            MonitorCallback monitorCallback = messageMonitor.onMessageIngested(command);
            Optional<MessageHandler<? super CommandMessage<?>>> optionalHandler = findCommandHandlerFor(command);
            if (optionalHandler.isPresent()) {
                CommandMessage<C> commandWithContext = spanFactory.propagateContext(command);
                CommandResultMessage<R> resultMessage = spanFactory.createHandleCommandSpan(commandWithContext).runSupplier(() -> {
                    log.debug("Handling command [{}]", commandWithContext.getCommandName());
                    InterceptorChain chain = new DefaultInterceptorChain<>(messageMessageTask, handlerInterceptors, optionalHandler.get());
                    try {
                        return asCommandResultMessage(chain.proceed());
                    } catch (Exception e) {
                        return asCommandResultMessage(e);
                    }
                });
                spanAwareCallback.onResult(command, resultMessage);
            } else {
                CommandHandlerNotFoundException exception = new CommandHandlerNotFoundException("No handler was subscribed for command [%s].".formatted(command.getCommandName()));
                monitorCallback.reportFailure(exception);
                spanAwareCallback.onResult(command, asCommandResultMessage(exception));
            }
        });

    }

    @Override
    public void batchDispatch(@Nonnull List<CommandMessage<?>> commandMessages) {

        BatchingMessageTask<CommandMessage<?>> batchingMessageTask = new BatchingMessageTask<>(commandMessages);
        batchingMessageTask.getOrComputeResource("_commands", s -> new ArrayList<>(commandMessages));
        batchingMessageTask.attachTransaction(transactionManager);

        final ResultMessage<? extends CommandResultMessage<?>> results = batchingMessageTask.executeWithResult((c) -> {
            log.info("执行命令,{}", c.getIdentifier());
            final CommandResultMessage<?>[] result = {null};
            dispatch(batchingMessageTask, c, (CommandCallback<Object, Object>) (commandMessage, commandResultMessage) -> {
                if (commandResultMessage.isExceptional()) {
                    final Throwable executeThrowable = commandResultMessage.optionalExceptionResult().orElse(null);
                    if (executeThrowable instanceof RuntimeException) {
                        throw (RuntimeException) executeThrowable;
                    }
                    throw new CommandExecutionException("执行批量命令时，其中[%s]命令执行异常".formatted(c.getCommandName()), executeThrowable);
                }
                result[0] = commandResultMessage;
            });
            return result[0];
        }, throwable -> true);
        if (results.isExceptional()) {
            final Throwable executeThrowable = results.optionalExceptionResult().orElse(null);
            log.error("批量执行命令异常", executeThrowable);
            if (executeThrowable instanceof CommandExecutionException commandExecutionException) {
                throw commandExecutionException;
            }
            if (executeThrowable == null) {
                throw new CommandExecutionException("批量执行命令异常：未知异常");
            } else {
                throw new CommandExecutionException("批量执行命令异常：" + executeThrowable.getMessage(), executeThrowable);
            }

        }

    }


    @Override
    public Registration subscribe(@Nonnull String commandName, @Nonnull CommandMessageHandler handler) {
        log.debug("Subscribing command with name [{}]", commandName);
        assertNonNull(handler, "handler may not be null");
        subscriptions.compute(commandName, (k, existingHandler) -> {
            if (existingHandler == null || existingHandler == handler) {
                return handler;
            } else {
                return duplicateCommandHandlerResolver.resolve(commandName, existingHandler, handler);
            }
        });
        return () -> subscriptions.remove(commandName, handler);
    }


    @Override
    public Registration registerHandlerInterceptor(@Nonnull MessageHandlerInterceptor<? super CommandMessage<?>> handlerInterceptor) {
        handlerInterceptors.add(handlerInterceptor);
        return () -> handlerInterceptors.remove(handlerInterceptor);
    }


    @Override
    public Registration registerDispatchInterceptor(@Nonnull MessageDispatchInterceptor<? super CommandMessage<?>> dispatchInterceptor) {
        dispatchInterceptors.add(dispatchInterceptor);
        return () -> dispatchInterceptors.remove(dispatchInterceptor);
    }


    public void setRollbackConfiguration(@Nonnull RollbackConfiguration rollbackConfiguration) {
        this.rollbackConfiguration = rollbackConfiguration;
    }
}
